package com.mingserve.raw.listener;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.mingserve.common.core.constant.CacheConstants;
import com.mingserve.common.core.constant.RawConstants;
import com.mingserve.common.core.utils.StringUtils;
import com.mingserve.common.mqtt.listener.MessageListener;
import com.mingserve.common.mqtt.utils.MqttUtils;
import com.mingserve.common.mqtt.utils.NetUtils;
import com.mingserve.common.redis.service.RedisService;
import com.mingserve.raw.domain.*;
import com.mingserve.raw.event.RefreshEvent;
import com.mingserve.raw.service.IDeviceService;
import com.mingserve.raw.service.IEquipmentService;
import com.mingserve.raw.service.IPointService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;

/**
 * 边缘服务器配置监听
 *
 * @author lvwshuai
 * @date 2024-02-01
 */
@RequiredArgsConstructor
@Slf4j
@Service
public class ConfigListener {
    private final MqttUtils mqttUtils;
    private final RedisService redisService;
    private final IEquipmentService equipmentService;
    private final IPointService pointService;
    private final IDeviceService deviceService;

    /**
     * 初始化Redis配置
     */
    public void init()
    {
        try {
            updateCache(CacheConstants.EDGE_SENSOR_LIST, deviceService::queryList);
            updateCache(CacheConstants.EDGE_EQUIP_LIST, equipmentService::queryList);
            updateCache(CacheConstants.EDGE_EQUIP_CONFIG_LIST, equipmentService::queryConfigList);
            updateCache(CacheConstants.EDGE_EQUIP_REPAIR_LIST, equipmentService::queryRepairList);
            updateCache(CacheConstants.EDGE_POINT_LIST, pointService::queryList);
            updateCache(CacheConstants.EDGE_POINT_CONFIG_LIST, pointService::queryConfigList);
            updateCache(CacheConstants.EDGE_POINT_BEARING_LIST, pointService::queryBearingList);
            updateCache(CacheConstants.EDGE_POINT_FAULT_LIST, pointService::queryFaultList);
            refreshSensorConfig();
            log.info("Init redis cache success");
        } catch (Exception e) {
            log.error("Init redis cache failed", e);
        }
    }

    /**
     * 刷新Redis缓存
     *
     * @param cacheKey
     * @param queryFunction
     * @param <T>
     *
     * @return 查询到的列表
     */
    private <T> List<T> updateCache(String cacheKey, Supplier<List<T>> queryFunction) {
        List<T> list = queryFunction.get();
        if (StringUtils.isNotEmpty(list)) {
            redisService.deleteObject(cacheKey);
            redisService.setCacheList(cacheKey, list);
            log.info("Refresh {} cache", cacheKey);
        }
        return list;
    }

    /**
     * 刷新测点缓存
     *
     * @param refreshEvent
     */
    @EventListener(RefreshEvent.class)
    private void refreshPointList(RefreshEvent refreshEvent)
    {
        if (refreshEvent.isRefreshPoint()) {
            updateCache(CacheConstants.EDGE_POINT_LIST, pointService::queryList);
            log.info("Refresh point list cache");
        }

        //  刷新设备采集配置及其对应的传感器配置
        if (StringUtils.isNotEmpty(refreshEvent.getEquipmentCode())) {
            // 刷新设备采集配置
            List<EquipmentConfig> equipConfigList = updateCache(CacheConstants.EDGE_EQUIP_CONFIG_LIST, equipmentService::queryConfigList);
            log.info("Refresh equip config cache");

            // 查询测点列表
            List<Point> pointList = redisService.getCacheList(CacheConstants.EDGE_POINT_LIST);
            if (StringUtils.isEmpty(pointList)) return;

            // 查询传感器列表
            List<Device> deviceList = updateCache(CacheConstants.EDGE_SENSOR_LIST, deviceService::queryList);
            // 预加载 deviceList，避免重复遍历
            Map<String, Device> deviceMap = deviceList.stream().collect(Collectors.toMap(Device::getDeviceCode, Function.identity(), (a, b) -> b));
            Map<String, String> configMap = new HashMap<>();

            // 查询最新采集配置
            String prefix = StringUtils.substring(refreshEvent.getEquipmentCode(), 0, refreshEvent.getEquipmentCode().length() - 3);
            List<EquipmentConfig> findList = equipConfigList.stream().filter(l -> l.getEquipmentCode().contains(prefix) && l.getEquipmentCode().length() < 28).toList();
            for (EquipmentConfig find : findList)
            {
                // 解析 JSON 配置
                JSONObject map = JSON.parseObject(find.getConfig());
                JSONObject config = Optional.ofNullable(map.getJSONObject("iphm_sensor_equipment")).orElse(map.getJSONObject("iphm_sensor_point"));
                if (config == null) continue;
                if (map.containsKey("iphm_sensor_point")) {
                    // 去掉设备级设备的开始使用标志
                    config.remove("StartUseFlag");
                }

                // 获取匹配的设备编码列表- 工作机组 + 设备级设备
                String currentPrefix = StringUtils.substring(find.getEquipmentCode(), 0, find.getEquipmentCode().length() - 3);
                List<String> deviceCodeList = pointList.stream().filter(l -> l.getPointCode().contains(currentPrefix) && StringUtils.isNotEmpty(l.getDeviceCode())).map(Point::getDeviceCode).toList();
                if (StringUtils.isEmpty(deviceCodeList)) continue;

                // 更新传感器配置
                deviceCodeList.forEach(deviceCode -> {
                    // 测点对应的传感器不存在，则新增传感器
                    Device device = deviceMap.getOrDefault(deviceCode, new Device());
                    JSONObject deviceConfig = Optional.ofNullable(JSON.parseObject(device.getConfig())).orElse(new JSONObject());
                    deviceConfig.putAll(config);
                    deviceConfig.put("SensorID", deviceCode);
                    device.setDeviceCode(deviceCode);
                    // 格式化配置数据类型
                    String json = JSON.toJSONString(DataFormatter.formatObj(deviceConfig));
                    device.setConfig(json);
                    deviceService.updateDevice(device);
                    // 更新缓存中的设备配置
                    deviceMap.put(deviceCode, device);
                    // 存储待变更的传感器配置
                    configMap.put(deviceCode, json);
                });
                redisService.deleteObject(CacheConstants.EDGE_SENSOR_LIST);
                redisService.setCacheList(CacheConstants.EDGE_SENSOR_LIST, new ArrayList<>(deviceMap.values()));
                log.info("Refresh sensor config under the equip {}", find.getEquipmentCode());
            }
            // 下发传感器配置
            configMap.forEach((key, value) -> {
                mqttUtils.publish(RawConstants.MQTT_DAQ, "cfg/" + key, value);
                log.info("Send sensor config {} {}", key, value);
            });
        }
    }

    /**
     * 订阅边缘服务器配置
     */
    public void subscribe()
    {
        String mac = NetUtils.getMacAddress();

        subscribeTopic(mac, "edge/sensor/", CacheConstants.EDGE_SENSOR_LIST, Device.class, deviceService::batchDevice);
        subscribeTopic(mac, "edge/equip/", CacheConstants.EDGE_EQUIP_LIST, Equipment.class, equipmentService::batchEquipment);
        subscribeTopic(mac, "edge/equipConfig/", CacheConstants.EDGE_EQUIP_CONFIG_LIST, EquipmentConfig.class, equipmentService::batchEquipmentConfig);
        subscribeTopic(mac, "edge/equipRepair/", CacheConstants.EDGE_EQUIP_REPAIR_LIST, EquipmentRepair.class, equipmentService::batchEquipmentRepair);
        subscribeTopic(mac, "edge/equipThreshold/", CacheConstants.EDGE_EQUIP_THRESHOLD_LIST, EquipmentThreshold.class, equipmentService::batchEquipmentThreshold);
        subscribeTopic(mac, "edge/point/", CacheConstants.EDGE_POINT_LIST, Point.class, pointService::batchPoint);
        subscribeTopic(mac, "edge/pointConfig/", CacheConstants.EDGE_POINT_CONFIG_LIST, PointConfig.class, pointService::batchPointConfig);
        subscribeTopic(mac, "edge/pointBearing/", CacheConstants.EDGE_POINT_BEARING_LIST, PointBearing.class, pointService::batchPointBearing);
        subscribeTopic(mac, "edge/pointFault/", CacheConstants.EDGE_POINT_FAULT_LIST, PointFault.class, pointService::batchPointFault);

        // 测点分组信息
        mqttUtils.subscribe(RawConstants.MQTT_MSW, "edge/pointGroup/" + mac, new MessageListener() {
            @Override
            public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
                redisService.setCacheObject(CacheConstants.EDGE_POINT_GROUP, new String(mqttMessage.getPayload()));
            }
        });
    }

    /**
     * 订阅MQTT消息并刷新缓存
     *
     * @param mac          设备的MAC地址
     * @param topicPrefix  MQTT主题前缀
     * @param cacheKey     Redis缓存键
     * @param clazz        消息解析的目标类
     * @param batchService 批量处理服务接口
     * @param <T>          泛型类型，表示消息解析后的对象类型
     */
    private <T> void subscribeTopic(String mac, String topicPrefix, String cacheKey, Class<T> clazz, BatchService<T> batchService) {
        String topic = topicPrefix + mac;
        mqttUtils.subscribe(RawConstants.MQTT_MSW, topic, new MessageListener() {
            @Override
            public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
                try {
                    log.info("Received {} MQTT {} msg", RawConstants.MQTT_MSW, topicPrefix);
                    List<T> list = JSON.parseArray(mqttMessage.getPayload(), clazz);
                    batchService.batch(list);
                    redisService.deleteObject(cacheKey);
                    if (!list.isEmpty()) {
                        redisService.setCacheList(cacheKey, list);
                    }
                    if (topic.contains("edge/sensor/")) refreshSensorConfig();
                    log.info("{} cache update completed", cacheKey);
                } catch (Exception ex) {
                    log.error("MQTT {} {} cache update failed", topic, cacheKey, ex);
                }
            }
        });
    }

    @FunctionalInterface
    interface BatchService<T> {
        void batch(List<T> list);
    }

    /**
     * 刷新传感器配置
     */
    void refreshSensorConfig()
    {
        List<Device> list = redisService.getCacheList(CacheConstants.EDGE_SENSOR_LIST);
        if (list != null) {
            list.forEach(device -> {
                if (StringUtils.isNotEmpty(device.getConfig())) {
                    log.info("Send sensor config {}", device.getDeviceCode());
                    mqttUtils.publishAsync(RawConstants.MQTT_DAQ, "cfg/" + device.getDeviceCode(), JSON.toJSONString(DataFormatter.formatObj(JSON.parseObject(device.getConfig()))));
                }
            });
        }
    }
}